/*-
 * Copyright 2012 Guram Dukashvili
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
//---------------------------------------------------------------------------
#include "sys.h"
#include "thread.h"
//---------------------------------------------------------------------------
struct thread * thread_new(void)
{
  struct thread * thr = salloc(sizeof(struct thread),NULL);
  if( thr != NULL )
    thread_init(thr);
  return thr;
}
//---------------------------------------------------------------------------
void thread_init(struct thread * thr)
{
  thr->exit_code = 0;
  semaphore_init(&thr->started);
  semaphore_init(&thr->restart);
  semaphore_init(&thr->finished);
  thr->terminate = false;
  thr->destroy = false;
#if __WIN32__
  thr->handle = NULL;
#elif HAVE_PTHREAD_H
  thr->handle = (pthread_t) NULL;
#endif
  thr->entry = NULL;
  thr->object = NULL;
  thr->param = 0;
  thr->pool = NULL;
  thr->pool_index = 0;
  thr->pool_used = false;
}
//---------------------------------------------------------------------------
void thread_destroy(struct thread * thr)
{
#if __WIN32__
  if( thr->handle != NULL ){
#elif HAVE_PTHREAD_H
  if( thr->handle != (pthread_t) NULL ){
#endif
    thr->terminate = true;
    thr->destroy = true;
    semaphore_post(&thr->restart);
    errno = 0;
#if __WIN32__
    if( thr->handle != NULL ){
      DWORD exit_code;
      BOOL r = GetExitCodeThread(thr->handle,&exit_code);
      if( r == 0 || exit_code == STILL_ACTIVE ){
        if( WaitForSingleObject(thr->handle,INFINITE) == WAIT_FAILED ){
          errno = EINVAL; //GetLastError() + errorOffset;
          perror(NULL);
#if COMPILE_RUNTIME_CHECKS
          fprintf(stderr,"undefined behavior in %s, %s, line %d\n",__PRETTY_FUNCTION__,__FILE__,__LINE__);
#endif
          abort();
        }
        GetExitCodeThread(thr->handle,&exit_code);
      }
      CloseHandle(thr->handle);
      thr->handle = NULL;
#elif HAVE_PTHREAD_H
    if( thr->handle != (pthread_t) NULL ){
      void * result;
      if( (errno = pthread_join(thr->handle,&result)) != 0 ){
        perror(NULL);
        abort();
      }
      thr->handle = (pthread_t) NULL;
#endif
    }
  }
}
//---------------------------------------------------------------------------
void thread_delete(struct thread * thr)
{
  thread_destroy(thr);
  sfree(thr);
}
//---------------------------------------------------------------------------
static
#if __WIN32__
DWORD entry(struct thread * thr)
#else
void * entry(struct thread * thr)
#endif
{
#if COMPILE_RUNTIME_CHECKS
  if( thr == NULL ){
    fprintf(stderr,"undefined behavior in %s, %s, line %d\n",__PRETTY_FUNCTION__,__FILE__,__LINE__);
    abort();
  }
  if( thr->entryf == NULL ){
    fprintf(stderr,"undefined behavior in %s, %s, line %d\n",__PRETTY_FUNCTION__,__FILE__,__LINE__);
    abort();
  }
#endif
#if HAVE_PTHREAD_H
  if( (errno = pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL)) != 0 ){
    perror(NULL);
    abort();
  }
  if( (errno = pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL)) != 0 ){
    perror(NULL);
    abort();
  }
#endif
#if __WIN32__
  thr->exit_code = STILL_ACTIVE;
#else
  thr->exit_code = 0;
#endif
  for(;;){
    semaphore_wait(&thr->restart);
    if( thr->destroy ) break;
    semaphore_post(&thr->started);
#if COMPILE_RUNTIME_CHECKS
    if( thr->entryf == NULL ){
      fprintf(stderr,"undefined behavior in %s, %s, line %d\n",__PRETTY_FUNCTION__,__FILE__,__LINE__);
      abort();
    }
#endif
    thr->entryf(thr,thr->object,thr->param);
#if __WIN32__
    if( thr->exit_code == STILL_ACTIVE )
      thr->exit_code = 0;
#endif
    semaphore_post(&thr->finished);
  }
#if __WIN32__
  return (DWORD) thr->exit_code;
#elif HAVE_PTHREAD_H
  return (void *) (intptr_t) thr->exit_code;
#endif
}
//---------------------------------------------------------------------------
static struct mutex runtime_mutex;
//---------------------------------------------------------------------------
void thread_init_runtime(void)
{
  mutex_init(&runtime_mutex);
}
//---------------------------------------------------------------------------
void thread_destroy_runtime(void)
{
  mutex_destroy(&runtime_mutex);
}
//---------------------------------------------------------------------------
void thread_start(struct thread * thr)
{
#if __WIN32__
  if( thr->handle == NULL ){
    mutex_acquire(&runtime_mutex);
    if( thr->handle == NULL ){
      thr->handle = CreateThread(NULL,256 * 1024,(LPTHREAD_START_ROUTINE) entry,thr,0/*CREATE_SUSPENDED*/,&thr->tid);
      if( thr->handle == NULL ){
        errno = EAGAIN; //GetLastError() + errorOffset;
        perror(NULL);
#if COMPILE_RUNTIME_CHECKS
        fprintf(stderr,"undefined behavior in %s, %s, line %d\n",__PRETTY_FUNCTION__,__FILE__,__LINE__);
#endif
        abort();
      }
#elif HAVE_PTHREAD_H
  if( thr->handle == (pthread_t) NULL ){
    mutex_acquire(&runtime_mutex);
    if( thr->handle == (pthread_t) NULL ){
      int err;
      pthread_attr_t attr;
      errno = pthread_attr_init(&attr);
      errno = pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_JOINABLE);
      errno = pthread_attr_setstacksize(&attr,256 * 1024);
      errno = pthread_create(&thr->handle,&attr,(void * (*)(void *)) entry,thr);
      err = errno;
      pthread_attr_destroy(&attr);
      errno = err;
      if( errno != 0 ){
        perror(NULL);
#if COMPILE_RUNTIME_CHECKS
        fprintf(stderr,"undefined behavior in %s, %s, line %d\n",__PRETTY_FUNCTION__,__FILE__,__LINE__);
#endif
        abort();
      }
#endif
    }
    mutex_release(&runtime_mutex);
  }
  thr->terminate = false;
  semaphore_post(&thr->restart);
  semaphore_wait(&thr->started);
}
//---------------------------------------------------------------------------
void thread_wait(struct thread * thr)
{
#if __WIN32__
  if( thr->handle != NULL )
#elif HAVE_PTHREAD_H
  if( thr->handle != (pthread_t) NULL ){
#endif
    semaphore_wait(&thr->finished);
  if( thr->pool != NULL ){
    mutex_acquire(&thr->pool->lock);
    vector_xchg_elements(&thr->pool->pool,thr->pool_index,thr->pool->pool.pointers.count - 1);
    thr->pool_used = false;
    mutex_release(&thr->pool->lock);
  }
}
//---------------------------------------------------------------------------
bool thread_active(struct thread * thr)
{
#if __WIN32__
  DWORD exit_code;
  return GetExitCodeThread(thr->handle,&exit_code) != 0 && exit_code == STILL_ACTIVE;
#elif HAVE_PTHREAD_H
  void * v;
  struct timespec abstime;
  abstime.tv_sec = 0;
  abstime.tv_nsec = 0;
  errno = pthread_timedjoin_np(thr->handle,&v,&abstime);
  return errno == ETIMEDOUT;
#else
  errno = ENOSYS;
  perror(NULL);
#if COMPILE_RUNTIME_CHECKS
  fprintf(stderr,"undefined behavior in %s, %s, line %d\n",__PRETTY_FUNCTION__,__FILE__,__LINE__);
#endif
  abort();
#endif
}
//---------------------------------------------------------------------------
struct thread_pool * thread_pool_new(void)
{
  struct thread_pool * pool = salloc(sizeof(struct thread_pool),NULL);
  if( pool != NULL )
    thread_pool_init(pool);
  return pool;
}
//---------------------------------------------------------------------------
static void thread_disable_copy(void)
{
  errno = ENOSYS;
  perror(NULL);
#if COMPILE_RUNTIME_CHECKS
  fprintf(stderr,"undefined behavior in %s, %s, line %d\n",__PRETTY_FUNCTION__,__FILE__,__LINE__);
#endif
  abort();
}
//---------------------------------------------------------------------------
void thread_pool_init(struct thread_pool * pool)
{
  mutex_init(&pool->lock);
  vector_init(&pool->pool);
  pool->pool.element_new = thread_new;
  pool->pool.element_copy = thread_disable_copy;
  pool->pool.element_delete = thread_delete;
  pool->max_size = 0;
  pool->cores = system_cores();
  pool->new_threads_count = 0;
  pool->reused_threads_count = 0;
}
//---------------------------------------------------------------------------
void thread_pool_destroy(struct thread_pool * pool)
{
  intptr_t i;

  mutex_acquire(&pool->lock);
  for( i = pool->pool.pointers.count - 1; i >= 0; i-- ){
    struct thread * thr = vector_get(&pool->pool,i);
    thr->destroy = true;
    semaphore_post(&thr->restart);
  }
  mutex_release(&pool->lock);

  vector_destroy(&pool->pool);
  mutex_destroy(&pool->lock);
}
//---------------------------------------------------------------------------
void thread_pool_delete(struct thread_pool * pool)
{
  thread_pool_destroy(pool);
  sfree(pool);
}
//---------------------------------------------------------------------------
void thread_pool_wait(struct thread_pool * pool)
{
  intptr_t i;
  struct thread * thr;

  //mutex_acquire(&pool->lock);
  for( i = pool->pool.pointers.count - 1; i >= 0; i-- ){
    thr = vector_get(&pool->pool,i);
    thr->terminate = true;
  }
  for( i = pool->pool.pointers.count - 1; i >= 0; i-- ){
    thr = vector_get(&pool->pool,i);
    thread_wait(thr);
  }
  //mutex_release(&pool->lock);
}
//---------------------------------------------------------------------------
struct thread * thread_pool_spare(struct thread_pool * pool,void * entry,void * object,uintptr_t param)
{
  intptr_t i;
  struct thread * thr = NULL, * thr2;

  mutex_acquire(&pool->lock);

  for( i = pool->pool.pointers.count - 1; i >= 0; i-- ){
    thr = vector_get(&pool->pool,i);
    if( !thr->pool_used ) break;
  }

  if( i < 0 ){
    thr = vector_append_get(&pool->pool);
    thr->pool = pool;
    thr->pool_index = pool->pool.pointers.count - 1;
    pool->new_threads_count++;
  }
  else {
    pool->reused_threads_count++;
  }

  thr->pool_used = true;

  if( pool->max_size > 0 ){
    for( i = pool->pool.pointers.count - 1; i >= 0; i-- ){
      if( pool->pool.pointers.count <= pool->max_size ) break;
      thr2 = vector_get(&pool->pool,i);
      if( thr2 != thr && !thr->pool_used ){
        thr->terminate = true;
        thr->destroy = true;
        vector_remove(&pool->pool,i);
      }
    }
    if( pool->pool.pointers.count > pool->max_size
      && pool->new_threads_count > pool->reused_threads_count ){
      pool->max_size <<= 1;
      pool->new_threads_count = pool->reused_threads_count = 0;
    }
    else if( pool->pool.pointers.count < pool->max_size
      && pool->new_threads_count < pool->reused_threads_count ){
      if( pool->max_size > 1 ) pool->max_size >>= 1;
      pool->new_threads_count = pool->reused_threads_count = 0;
    }
  }

  thr->entry = entry;
  thr->object = object;
  thr->param = param;

  mutex_release(&pool->lock);

  thread_start(thr);

  return thr;
}
//---------------------------------------------------------------------------
struct shared_data {
  struct mutex lock;
  struct rwlock rwlock;

  uintptr_t rshared;
  uintptr_t wshared;
  uint64_t rtime;
  uint64_t wtime;
  uintptr_t it;

  struct semaphore go;
};

static __inline__ void shared_data_init(struct shared_data * data)
{
  mutex_init(&data->lock);
  rwlock_init(&data->rwlock);
  data->rshared = data->wshared = 0;
  data->rtime = data->wtime = 0;
  data->it = 0;
  semaphore_init(&data->go);
}

static __inline__ void shared_data_destroy(struct shared_data * data)
{
  rwlock_destroy(&data->rwlock);
  mutex_destroy(&data->lock);
}

static void thread_locking_test_entry(struct thread * thr,struct shared_data * data,enum RWLOCK_TYPE lock_type)
{
  intptr_t i;
  uint64_t sti;

  thr = thr;
  semaphore_wait(&data->go);

  sti = get_time();

  for( i = data->it - 1; i >= 0; i-- ){
    rwlock_acquire(&data->rwlock,lock_type);
    if( lock_type == RWL_READ ){
      data->rshared++;
    }
    else if( lock_type == RWL_WRITE ){
      data->wshared++;
    }
    rwlock_release(&data->rwlock,lock_type);
  }

  sti = get_time() - sti + 1;

  mutex_acquire(&data->lock);
  if( lock_type == RWL_READ ){
    data->rtime += sti;
  }
  else if( lock_type == RWL_WRITE ){
    data->wtime += sti;
  }
  mutex_release(&data->lock);
}
//---------------------------------------------------------------------------
void thread_locking_test(uintptr_t rthreads,uintptr_t wthreads,uintptr_t it)
{
  intptr_t i;
  double r, w;
  struct shared_data data;
  struct thread_pool pool;

  fprintf(stderr,"-=< thread locking test started >=-\n");

  shared_data_init(&data);
  thread_pool_init(&pool);

  data.it = it;

  for( i = rthreads + wthreads - 1; i >= 0; i-- )
    thread_pool_spare(&pool,
      thread_locking_test_entry,
      &data,(uintptr_t) i < rthreads ? RWL_READ : RWL_WRITE);

  semaphore_post_v(&data.go,rthreads + wthreads);
  thread_pool_wait(&pool);

  r = (double) (rthreads * it) * 1000000u / data.rtime;
  w = (double) (wthreads * it) * 1000000u / data.wtime;

  fprintf(stderr,"rthreads %" PRIuPTR", %.4f rlps\n",rthreads,r);
  fprintf(stderr,"wthreads %" PRIuPTR", %.4f wlps\n",wthreads,w);
  fprintf(stderr,"%.4f(%.4f) r/w ratio\n",(double) rthreads / wthreads,r / w);
  fprintf(stderr,wthreads * it == data.wshared
    ? "-=< thread locking test finished successfuly >=-\n"
    : "-=< thread locking test failed >=-\n"
  );

  thread_pool_destroy(&pool);
  shared_data_destroy(&data);
}
//---------------------------------------------------------------------------
